import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CharsetEncoder;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

//基于nio非阻塞方式的聊天室服务端
public class Server {
	private final int port;
	private Selector ac_selector, read_selector, write_selector;
	private ServerSocketChannel ssc = null;
	private CharsetEncoder encoder = Charset.forName("utf-8").newEncoder();
	private CharsetDecoder decoder = Charset.forName("utf-8").newDecoder();
	private ConcurrentHashMap<SelectionKey,Queue> msgtable = new ConcurrentHashMap<SelectionKey,Queue>();
	private String seperator = System.getProperty("line.separator");
	public Server(int port){
		this.port = port;
		try {
			ssc = ServerSocketChannel.open();  //打开通道
			ssc.configureBlocking(false);  //设置为非阻塞
			ssc.socket().bind(new InetSocketAddress(port));
			
			ac_selector = Selector.open();   //创建选择器
			read_selector = Selector.open();
			
			ssc.register(ac_selector, SelectionKey.OP_ACCEPT);  //监听连接事件
		} catch (IOException e) {
			// TODO Auto-generated catch block
			System.out.println("server start failed:"+e.getMessage());
		}
		
	}
	
	private int genKey(){
		return new Random().nextInt();
	}
	
	public void start(){
		//启动读取监听线程
		new Thread(new Runnable(){

			@Override
			public void run() {
				
					while (true) {
						try{
							
							//if(read_selector.selectNow()==0)continue;
							if(read_selector.select(5)==0)continue;
							Iterator itr = read_selector.selectedKeys().iterator();
							while(itr.hasNext()){
								SelectionKey key = (SelectionKey) itr.next();
								itr.remove();  //移除该事件自己
								if(key.isReadable()){   //处理读数据
									//System.out.println("read ready");
									receiveMsg(key);
									//System.out.println("having read a msg");
								}
								if(key.isWritable()){   //处理写数据
									//System.out.println("write ready");
									writeMsg(key);
									
									
								}
							}
						} catch(CancelledKeyException ce){
							System.out.println("a key has been cancelled");
						}catch (Exception e) {
							// TODO Auto-generated catch block
							System.out.println("read listener error:"+e.getMessage());
							e.printStackTrace();
						}
					}
				
			}
			
		}).start();
		//启动写监听线程
		/*new Thread(new Runnable(){

			@Override
			public void run() {
				try {
					while (true) {
						if(write_selector.selectNow()==0)continue;
						Iterator itr = write_selector.selectedKeys().iterator();
						while(itr.hasNext()){
							SelectionKey key = (SelectionKey) itr.next();
							itr.remove();  //移除该事件自己
							
						}
					}
				} catch (IOException e) {
					// TODO Auto-generated catch block
					System.out.println("read error:"+e.getMessage());
				}
			}
			
		}).start();*/
		//主线程负责监听连接
		try {
			while(true){
				//System.out.println("block in ac select");
				ac_selector.select();  //获取发生事件的通道，此方法阻塞
				
				Iterator itr = ac_selector.selectedKeys().iterator();
				while(itr.hasNext()){
					SelectionKey key = (SelectionKey) itr.next();
					ServerSocketChannel subssc = (ServerSocketChannel) key.channel();
					itr.remove();  //移除该事件自己
					if(key.isAcceptable()){  //连接事件
						SocketChannel sc = subssc.accept();
						//System.out.println("get a client"+sc);
						
						if(sc!=null){
							
							
							sc.configureBlocking(false);   //设置连接异步
							
							//read_selector.wakeup();  //中止阻塞，以便注册
							SelectionKey newkey = sc.register(read_selector.wakeup(), SelectionKey.OP_READ | SelectionKey.OP_WRITE);   //监听读写事件
							
							//为每个连接创建一个消息队列
							msgtable.put(newkey, new ArrayBlockingQueue<String>(200));
							//sc.register(write_selector,  SelectionKey.OP_WRITE).attach(id); 
						}
						
						
						
					}
					
					
				}
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
	
	private void receiveMsg(SelectionKey key){
		SocketChannel sc = (SocketChannel)key.channel();
		ByteBuffer buffer = ByteBuffer.allocate(1024);
		CharBuffer cb = null;
		StringBuilder sb = new StringBuilder();
		try {
			int count = 0; 
			while( (count = sc.read(buffer)) > 0){ //读入数据
				//System.out.println("get a msg");
				buffer.flip();   //读状态
				cb = decoder.decode(buffer);
				sb.append(cb.array());
				//检测是否得到字符结束标志
				int end = 0;
				while((end = sb.indexOf(seperator)) > -1){
					
					//收到一个消息，向其他客户端发出
					String msg = sb.substring(0, end);
					broadCastMsg(key,msg+seperator);
					sb.delete(0, end+seperator.length());
					sb.trimToSize();
				}
				buffer.clear();  //清缓存
			}
			
		} catch (IOException e) {
			// TODO Auto-generated catch block
			System.out.println("read socket error:"+e.getMessage());
			System.out.println("remove a client");
			msgtable.remove(key);
			try {
				sc.close();
			} catch (IOException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
			key.cancel();
		}
	}
	
	
	private void writeMsg(SelectionKey key){
		//查看自己的消息队列有没有未发消息
		
		String msg = null;
		while((msg = (String)(msgtable.get(key).poll())) != null){
			SocketChannel sc = (SocketChannel)key.channel();
			try {
				ByteBuffer buffer = encoder.encode(CharBuffer.wrap(msg));
				if(sc != null){
					sc.write(buffer);
				}
			} catch (Exception e) {
				// TODO Auto-generated catch block
				System.out.println("write socket error:"+e.getMessage());
				System.out.println("remove a client");
				msgtable.remove(key);
				try {
					sc.close();
				} catch (IOException e1) {
					// TODO Auto-generated catch block
					e1.printStackTrace();
				}
			}
			
		}
		key.interestOps(SelectionKey.OP_READ);  //发送完了后只关心读事件
		key.selector().wakeup();
	}
	
	private void broadCastMsg(SelectionKey fromid,String msg){
		//System.out.println("borad a msg  "+msg);
		//向所有其他客户端的消息队列中放入消息
		Iterator itr = msgtable.keySet().iterator();
		while(itr.hasNext()){
			SelectionKey id = (SelectionKey)itr.next();
			if(!id.equals(fromid)){
				msgtable.get(id).offer(msg);
				id.interestOps(SelectionKey.OP_WRITE | SelectionKey.OP_READ);  //让各个客户端通道关心读写事件 
				id.selector().wakeup();
			}
		}
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		new Server(9901).start();

	}

}
